package pkg

import (
	"github.com/streadway/amqp"
)

var (
	RabbitMqConn *amqp.Connection
	err          error
)

func InitRabbitMq() {
	RabbitMqConn, err = amqp.Dial("amqp://guest:guest@localhost:5672/")
	if err != nil {
		panic(err)
		return
	}
}

// 简单模式 生产者
func SimpleProducer(queueName string, body string) {
	channel, err := RabbitMqConn.Channel()
	if err != nil {
		panic(err)
		return
	}
	declare, err := channel.QueueDeclare(
		queueName, // name of the queue
		true,      // durable
		false,     // delete when unused
		false,     // exclusive
		false,     // noWait
		nil,       // arguments
	)
	if err != nil {
		return
	}
	err = channel.Publish(
		"",           // publish to an exchange
		declare.Name, // routing to 0 or more queues
		false,        // mandatory
		false,        // immediate
		amqp.Publishing{
			Headers:         amqp.Table{},
			ContentType:     "text/plain",
			ContentEncoding: "",
			Body:            []byte(body),
			DeliveryMode:    amqp.Persistent, // 1=non-persistent, 2=persistent
			Priority:        0,               // 0-9
			// a bunch of application/implementation-specific fields
		},
	)
	if err != nil {
		return
	}
}

type callBulk func(str string)

// 简单模式 消费者
func SimpleConsumer(queueName string, callBulk callBulk) {
	channel, err := RabbitMqConn.Channel()
	if err != nil {
		panic(err)
		return
	}
	declare, err := channel.QueueDeclare(
		queueName, // name of the queue
		true,      // durable
		false,     // delete when unused
		false,     // exclusive
		false,     // noWait
		nil,       // arguments
	)
	if err != nil {
		return
	}
	consume, err := channel.Consume(
		declare.Name, // name
		"",           // consumerTag,
		false,        // noAck
		false,        // exclusive
		false,        // noLocal
		false,        // noWait
		nil,          // arguments
	)
	if err != nil {
		return
	}
	for d := range consume {
		callBulk(string(d.Body))
		d.Ack(false)
	}
}
